package com.jorado.pulsar.client.ha;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.ServiceUrlProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZookeeperServiceUrlProvider implements ServiceUrlProvider {

    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperServiceUrlProvider.class);

    private PulsarClient pulsarClient;
    private String serviceUrl;
    private CuratorFramework client;
    private NodeCache nodeCache;
    private String path;

    @Override
    public void initialize(PulsarClient pulsarClient) {
        this.pulsarClient = pulsarClient;
        try {
            nodeCache = new NodeCache(this.client, path);
            nodeCache.start(false);
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public synchronized void nodeChanged() throws Exception {
                    String serviceUrlFromZK = new String(nodeCache.getCurrentData().getData());
                    if (!serviceUrlFromZK.equals(getServiceUrl())) {
                        setServiceUrl(serviceUrlFromZK);
                        pulsarClient.updateServiceUrl(serviceUrl);
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static Builder builder() {
        return new Builder();
    }

    @Override
    public String getServiceUrl() {
        return serviceUrl;
    }

    public ZookeeperServiceUrlProvider(String zookeeperUrl, String path) {
        try {
            this.path = path;
            initClient(zookeeperUrl);
            setServiceUrl(new String(this.client.getData().forPath(path)));
        } catch (Exception e) {
            LOG.error("ServiceProviderTest start fail", e);
        }
    }

    public PulsarClient getPulsarClient() {
        return pulsarClient;
    }

    private void setServiceUrl(String serviceUrl) {
        this.serviceUrl = serviceUrl;
    }

    private void initClient(String zookeeperUrl) {
        this.client = CuratorFrameworkFactory.builder()
                .connectString(zookeeperUrl)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .sessionTimeoutMs(10000)
                .connectionTimeoutMs(10000)
                .build();
        this.client.start();
    }

    public static class Builder {

        private String zookeeperUrl;
        private String path;

        public Builder zookeeperUrl(String zookeeperUrl) {
            this.zookeeperUrl = zookeeperUrl;
            return this;
        }

        public Builder path(String path) {
            this.path = path;
            return this;
        }

        public ZookeeperServiceUrlProvider build() {
            return new ZookeeperServiceUrlProvider(zookeeperUrl, path);
        }
    }

}


